#include "in.h"

size_t CalcMessageDataUnitSize( uint8_t topic_name_len , uint16_t tags_len , uint32_t msg_data_len )
{
	return sizeof(uint64_t) + sizeof(uint8_t) + topic_name_len + sizeof(uint16_t) + tags_len + sizeof(uint32_t) + msg_data_len ;
}

uint64_t MemWrite_MessageDataUnit( char *unit_offset_ptr , uint64_t msg_offset , uint8_t topic_name_name_len , char *topic_name , uint16_t tags_len , char *tags , uint32_t msg_data_len , char *msg_data )
{
	char		*p = unit_offset_ptr ;
	uint64_t	pack_len ;
	
	*(uint64_t*)p = hton64(msg_offset) ;
	p += sizeof(uint64_t) ;
	
	*(uint8_t*)p = topic_name_name_len ;
	p += sizeof(uint8_t) ;
	
	memcpy( p , topic_name , topic_name_name_len );
	p += topic_name_name_len ;
	
	*(uint16_t*)p = hton16(tags_len) ;
	p += sizeof(uint16_t) ;
	
	memcpy( p , tags , tags_len );
	p += tags_len ;
	
	*(uint32_t*)p = hton32(msg_data_len) ;
	p += sizeof(uint32_t) ;
	
	memcpy( p , msg_data , msg_data_len );
	p += msg_data_len ;
	
	*(uint64_t*)p = hton64(0) ;
	
	pack_len = p - unit_offset_ptr ;
	
	return pack_len;
}

uint64_t MemRead_MessageDataUnit( char *unit_offset_ptr , uint64_t *p_msg_offset , uint8_t *p_topic_name_len , char **pp_topic_name , uint16_t *p_tags_len , char **pp_tags , uint32_t *p_msg_data_len , char **pp_msg_data )
{
	char		*p = unit_offset_ptr ;
	uint64_t	msg_offset ;
	uint8_t		topic_name_len ;
	uint16_t	tags_len ;
	uint32_t	msg_data_len ;
	uint64_t	unpack_len ;
	
	msg_offset = ntoh64(*(uint64_t*)p) ;
	if( p_msg_offset )
		(*p_msg_offset) = msg_offset ;
	if( msg_offset == 0 || msg_offset == UINT64_MAX )
		return msg_offset;
	p += sizeof(uint64_t) ;
	
	topic_name_len = *(uint8_t*)p ;
	if( p_topic_name_len )
		(*p_topic_name_len) = topic_name_len ;
	p += sizeof(uint8_t) ;
	
	if( pp_topic_name )
		(*pp_topic_name) = p ;
	p += topic_name_len ;
	
	tags_len = ntoh16(*(uint16_t*)p) ;
	if( p_tags_len )
		(*p_tags_len) = tags_len ;
	p += sizeof(uint16_t) ;
	
	if( pp_tags )
		(*pp_tags) = p ;
	p += tags_len ;
	
	msg_data_len = ntoh32(*(uint32_t*)p) ;
	if( p_msg_data_len )
		(*p_msg_data_len) = msg_data_len ;
	p += sizeof(uint32_t) ;
	
	if( pp_msg_data )
		(*pp_msg_data) = p ;
	p += msg_data_len ;
	
	unpack_len = p - unit_offset_ptr ;
	
	return unpack_len;
}

int CreateMessageDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , size_t begin_offset , struct MQFileMapping **pp_msg_file , size_t *p_file_size )
{
	struct MQFileMapping	*msg_file = NULL ;
	int			file_fd ;
	off_t			seek_file_offset ;
	ssize_t			write_file_len ;
	
	int			nret = 0 ;
	
	msg_file = (struct MQFileMapping *)malloc( sizeof(struct MQFileMapping) ) ;
	if( msg_file == NULL )
	{
		ERRORLOG( "malloc failed , errno[%d]" , errno )
		if( comm_data )
			MEMPACK_RESULT( return -11 , MQSDK_ERROR_MALLOC , comm_data )
		return -1;
	}
	memset( msg_file , 0x00 , sizeof(struct MQFileMapping) );
	
	memset( msg_file->path_file_name , 0x00 , sizeof(msg_file->path_file_name) );
	snprintf( msg_file->path_file_name , sizeof(msg_file->path_file_name)-1 , "%s/%.*s/%020zu.cmqm" , broker->data_path , store->store_name_len,store->store_name , begin_offset );
	file_fd = open( msg_file->path_file_name , O_CREAT|O_EXCL|O_RDWR|O_TRUNC , 00777 ) ;
	if( file_fd == -1 )
	{
		free( msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -12 , MQSDK_ERROR_CREATE_MSG_DATA_FILE , comm_data )
		return -2;
	}
	else
	{
		DEBUGLOG( "open[%s] ok" , msg_file->path_file_name )
	}
	
	msg_file->begin_offset = begin_offset ;
	msg_file->file_size = broker->msg_file_size ;
	
	seek_file_offset = lseek( file_fd , msg_file->file_size-1 , SEEK_SET ) ;
	if( seek_file_offset == -1 )
	{
		ERRORLOG( "lseek[%s] offset[%zu] failed , errno[%d]" , msg_file->path_file_name , msg_file->file_size-1 , errno )
		close( file_fd );
		unlink( msg_file->path_file_name );
		free( msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -13 , MQSDK_ERROR_LSEEK_MSG_DATA_FILE , comm_data )
		return -3;
	}
	else
	{
		DEBUGLOG( "lseek[%s] offset[%zu] ok" , msg_file->path_file_name , msg_file->file_size-1 )
	}
	
	write_file_len = write( file_fd , "\0" , 1 ) ;
	if( write_file_len == -1 )
	{
		ERRORLOG( "write[%s] failed , errno[%d]" , msg_file->path_file_name , errno )
		close( file_fd );
		unlink( msg_file->path_file_name );
		free( msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -14 , MQSDK_ERROR_LWRITE_MSG_DATA_FILE , comm_data )
		return -4;
	}
	else
	{
		DEBUGLOG( "write[%s] ok" , msg_file->path_file_name )
	}
	
	msg_file->mmap_size = msg_file->file_size ;
	msg_file->file_mmap = mmap( NULL , msg_file->file_size , PROT_READ|PROT_WRITE , MAP_SHARED , file_fd , 0 ) ;
	if( msg_file->file_mmap == (void*)(-1) )
	{
		ERRORLOG( "mmap[%s] failed , errno[%d]" , msg_file->path_file_name , errno )
		close( file_fd );
		unlink( msg_file->path_file_name );
		free( msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -15 , MQSDK_ERROR_LWRITE_MSG_DATA_FILE , comm_data )
		return -5;
	}
	else
	{
		DEBUGLOG( "mmap[%s] size[%zu] ok , addr[0x%X]" , msg_file->path_file_name , msg_file->file_size , msg_file->file_mmap )
	}
	
	close( file_fd );
	
	memset( msg_file->file_mmap , 0x00 , msg_file->file_size );
	msg_file->fill_ptr = msg_file->file_mmap ;
	msg_file->over_end_ptr = msg_file->file_mmap + msg_file->file_size ;
	DEBUGLOG( "msg_file->over_end_ptr[0x%X]" , msg_file->over_end_ptr )
	
	DEBUGLOG( "write msg file head" )
	memcpy( msg_file->fill_ptr+MQ_MSG_FILE_HEADER_FORMAT_MAGIC_OFFSET , MQ_MSG_FILE_HEADER_FORMAT_MAGIC_VALUE , MQ_MSG_FILE_HEADER_FORMAT_MAGIC_LENGTH );
	*(msg_file->fill_ptr+MQ_MSG_FILE_HEADER_FORMAT_VERSION_OFFSET) = MQ_MSG_FILE_RESTORE_FORMAT_VERSION_1 ;
	*((uint64_t*)(msg_file->fill_ptr+MQ_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_OFFSET)) = hton64(begin_offset) ;
	*((uint64_t*)(msg_file->fill_ptr+MQ_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_OFFSET)) = hton64(msg_file->file_size) ;
	*(msg_file->fill_ptr+MQ_MSG_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET) = MQ_MSG_FILE_HEADER_FORMAT_NOT_CLEAN ;
	msg_file->fill_ptr += MQ_MSG_FILE_HEAD_SIZE ;
	DEBUGLOG( "msg_file->fill_ptr[0x%X]" , msg_file->fill_ptr )
	
	store->max_msg_offset += MQ_MSG_FILE_HEAD_SIZE ;
	DEBUGLOG( "store->max_msg_offset[%zu]" , store->max_msg_offset )
	
	INIT_LIST_HEAD( & (msg_file->subscribe_session_list) );
	
	DEBUGLOG( "add msg_file[%s] to store's msg_file_list" , msg_file->path_file_name )
	list_add_tail( & (msg_file->file_list_node) , & (store->msg_file_list) );
	store->msg_file_list_nodes_count++;
	
	if( pp_msg_file )
		(*pp_msg_file) = msg_file ;
	if( p_file_size )
		(*p_file_size) = msg_file->file_size ;
	return 0;
}

int LoadMessageDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , size_t begin_offset , char *filename , struct MQFileMapping **pp_msg_file , size_t *p_file_size )
{
	struct MQFileMapping	*msg_file = NULL ;
	int			file_fd ;
	
	int			nret = 0 ;
	
	msg_file = (struct MQFileMapping *)malloc( sizeof(struct MQFileMapping) ) ;
	if( msg_file == NULL )
	{
		ERRORLOG( "malloc failed , errno[%d]" , errno )
		if( comm_data )
			MEMPACK_RESULT( return -11 , MQSDK_ERROR_MALLOC , comm_data )
		return -1;
	}
	memset( msg_file , 0x00 , sizeof(struct MQFileMapping) );
	
	memset( msg_file->path_file_name , 0x00 , sizeof(msg_file->path_file_name) );
	if( filename == NULL )
		snprintf( msg_file->path_file_name , sizeof(msg_file->path_file_name)-1 , "%s/%.*s/%020zu.cmqm" , broker->data_path , store->store_name_len,store->store_name , begin_offset );
	else
		strncpy( msg_file->path_file_name , filename , sizeof(msg_file->path_file_name)-1 );
	DEBUGLOG( "try to open[%s]" , msg_file->path_file_name )
	file_fd = open( msg_file->path_file_name , O_RDWR , 00777 ) ;
	if( file_fd == -1 )
	{
		DEBUGLOG( "open[%s] not exist" , msg_file->path_file_name )
		free( msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -12 , MQSDK_ERROR_CREATE_MSG_DATA_FILE , comm_data )
		return -2;
	}
	else
	{
		DEBUGLOG( "open[%s] ok" , msg_file->path_file_name )
	}
	
	msg_file->mmap_size = GetFileSize( NULL , file_fd ) ;
	if( msg_file->mmap_size < MQ_MSG_FILE_HEAD_SIZE )
	{
		ERRORLOG( "file size[%z] invalid" , msg_file->mmap_size )
		close( file_fd );
		unlink( msg_file->path_file_name );
		free( msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -13 , MQSDK_ERROR_MSG_DATA_FILE_SIZE_INVALID , comm_data )
		return -3;
	}
	msg_file->file_mmap = mmap( NULL , msg_file->mmap_size , PROT_READ|PROT_WRITE , MAP_SHARED , file_fd , 0 ) ;
	if( msg_file->file_mmap == (void*)(-1) )
	{
		ERRORLOG( "mmap[%s] failed , errno[%d]" , msg_file->path_file_name , errno )
		close( file_fd );
		unlink( msg_file->path_file_name );
		free( msg_file );
		if( comm_data )
			MEMPACK_RESULT( return -13 , MQSDK_ERROR_LWRITE_MSG_DATA_FILE , comm_data )
		return -3;
	}
	else
	{
		if( filename == NULL )
			msg_file->begin_offset = begin_offset ;
		else
			msg_file->begin_offset = *((uint64_t*)(msg_file->file_mmap+5)) ;
		msg_file->file_size = ntoh64(*((uint64_t*)(msg_file->file_mmap+13))) ;
		DEBUGLOG( "mmap[%s] mmap_size[%zu] file_size[%zu] ok , addr[0x%X]" , msg_file->path_file_name , msg_file->mmap_size , msg_file->file_size , msg_file->file_mmap )
	}
	
	if( p_file_size )
	{
		(*p_file_size) = msg_file->file_size ;
		DEBUGLOG( "(*p_file_size)[%zu]" , (*p_file_size) )
	}
	
	close( file_fd );
	
	if( *(msg_file->file_mmap+MQ_MSG_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET) == MQ_MSG_FILE_HEADER_FORMAT_NOT_CLEAN )
	{
		msg_file->over_end_ptr = msg_file->file_mmap + msg_file->mmap_size ;
		msg_file->fill_ptr = msg_file->file_mmap + MQ_MSG_FILE_HEAD_SIZE ;
		
		INIT_LIST_HEAD( & (msg_file->subscribe_session_list) );
		
		if( store )
		{
			INFOLOG( "add msg_file[%s] to store's msg_file_list" , msg_file->path_file_name )
			list_add_tail( & (msg_file->file_list_node) , & (store->msg_file_list) );
			store->msg_file_list_nodes_count++;
		}
	}
	else
	{
		munmap( msg_file->file_mmap , msg_file->mmap_size );
		free( msg_file ); msg_file = NULL ;
	}
	
	if( pp_msg_file )
	{
		(*pp_msg_file) = msg_file ;
		DEBUGLOG( "ret (*pp_msg_file)[%p]" , (*pp_msg_file) )
	}
	
	return 0;
}

int LoadAllMessageDataFiles( struct MQBroker *broker , struct MQStore *store )
{
	size_t			begin_size ;
	struct MQFileMapping	*msg_file = NULL ;
	size_t			file_size ;
	uint64_t		read_len ;
	
	int			nret = 0 ;
	
	begin_size = 0 ;
	for( ; ; )
	{
		nret = LoadMessageDataFile( NULL , broker , store , begin_size , NULL , & msg_file , & file_size ) ;
		if( nret == -2 )
		{
			break;
		}
		else if( nret )
		{
			ERRORLOG( "LoadMessageDataFile failed" )
			break;
		}
		else
		{
			INFOLOG( "LoadMessageDataFile ok , begin_size[%zu] file_size[%zu]" , begin_size , file_size )
		}
		
		for( ; ; )
		{
			read_len = MemRead_MessageDataUnit( msg_file->fill_ptr , NULL , NULL , NULL , NULL , NULL , NULL , NULL ) ;
			if( read_len == 0 || read_len == UINT64_MAX )
			{
				if( store )
					store->max_msg_offset = msg_file->fill_ptr - msg_file->file_mmap + msg_file->begin_offset ;
				break;
			}
			else
			{
				msg_file->fill_ptr += read_len ;
				if( store )
					store->max_msg_offset += read_len ;
			}
		}
		
		begin_size += file_size ;
	}
	
	return 0;
}

int UnloadMessageDataFile( struct MQBroker *broker , struct MQStore *store , struct MQFileMapping *msg_file )
{
	munmap( msg_file->file_mmap , msg_file->file_size );
	
	list_del( & (msg_file->file_list_node) );
	store->msg_file_list_nodes_count--;
	
	free( msg_file );
	
	return 0;
}

int CleanMessageDataFile( struct MQBroker *broker , struct MQStore *store , struct MQFileMapping *msg_file )
{
	char		path_file_name[ PATH_MAX ] ;
	
	int		nret = 0 ;
	
	strcpy( path_file_name , msg_file->path_file_name );
	
	DEBUGLOG( "clean msg_file[%s]" , path_file_name )
	*(msg_file->file_mmap+MQ_MSG_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET) = MQ_MSG_FILE_HEADER_FORMAT_CLEANNED ;
	
	DEBUGLOG( "UnloadMessageDataFile ... store[%s] msg_file[%s]" , store->store_name , path_file_name )
	nret = UnloadMessageDataFile( broker , store , msg_file ) ;
	if( nret )
	{
		ERRORLOG( "UnloadMessageDataFile failed[%d]" , nret )
		return nret;
	}
	else
	{
		DEBUGLOG( "UnloadMessageDataFile ok" )
	}
	
	DEBUGLOG( "truncate msg_file[%s]" , path_file_name )
	truncate( path_file_name , MQ_MSG_FILE_HEAD_SIZE );
	
	return 0;
}

int UnloadAllMessageDataFiles( struct MQBroker *broker , struct MQStore *store )
{
	struct MQFileMapping	*msg_file = NULL ;
	struct MQFileMapping	*next_msg_file = NULL ;
	
	list_for_each_entry_safe( msg_file , next_msg_file , & (store->msg_file_list) , struct MQFileMapping , file_list_node )
	{
		UnloadMessageDataFile( broker , store , msg_file );
	}
	
	return 0;
}

